package com.neusoft.bd.myflink.step06;

import com.neusoft.bd.myflink.entity.WaterSensor;
import com.neusoft.bd.myflink.source.WaterSensorSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.kafka.common.protocol.types.Field;

import java.util.Date;
import java.util.Random;

public class WindoDemo01 {

    public static void main(String[] args) throws Exception {
        final Random r = new Random(System.currentTimeMillis());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream =
                env.fromSource(
                        WaterSensorSource.waterSensorDataStreamSource(),
                        WatermarkStrategy.noWatermarks(),
                        "Generator Source");
        stream.keyBy(ws -> ws.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(19)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {

                    @Override
                    public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        System.out.println(new Date(context.window().getStart()).toString() + "----> " + context.window().getEnd());
                        final Tuple2<Integer, Integer> tuple2 = Tuple2.of(0, 0);
                        elements.forEach(e -> {
                            tuple2.f0++;
                            tuple2.f1 += e.getVc();

                        });
                        String str = key + " : " + tuple2.f1 / tuple2.f0;
                        out.collect(str);

                    }
                })
                .print();


        env.execute();
    }

    public static void main1(String[] args) throws Exception {
        final Random r = new Random(System.currentTimeMillis());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream =
                env.fromSource(
                        WaterSensorSource.waterSensorDataStreamSource(),
                        WatermarkStrategy.noWatermarks(),
                        "Generator Source");
        stream.keyBy(ws -> ws.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<WaterSensor>() {
                    @Override
                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                        System.out.println("调用reduce方法，之前的结果:" + value1 + ",现在来的数据:" + value2);
                        return new WaterSensor(value1.getId(), System.currentTimeMillis(), value1.getVc() + value2.getVc());
                    }
                }).print();
        env.execute();
    }
}
